"""
Case Type   : Logical_Replication
Case Name   : 验证pg_logical_slot_peek_binary_changes函数slotname参数
Create At   : 2024/1/5
Owner       : xiaqi14
Description :
    1.修改参数
    2.配置初始用户白名单
    3.创建逻辑复制槽
    4.在主机上执行此函数，验证slotname参数
    5.在备机上执行此函数，验证slotname参数
    6.删除逻辑复制槽
    7.恢复参数，恢复pg_hba.conf文件
Expect      :
    1.修改参数成功
    2.配置成功
    3.创建成功
    4.只有存在的slotname执行成功，其他的合理报错
    5.只有存在的slotname执行成功，其他的合理报错
    6.删除成功
    7.恢复成功
History     :
"""
import os
import unittest
from yat.test import Node
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant

primary_sh = CommonSH('PrimaryDbUser')


@unittest.skipIf(1 == primary_sh.get_node_num(), '单机环境不执行')
class LogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.primary_node = Node('PrimaryDbUser')
        self.sh_standby = CommonSH('Standby1DbUser')
        self.constant = Constant()
        self.common = Common()
        self.slot_name_1 = 'slot_logical_replication_0070_1'
        self.slot_name_2 = 'slot_logical_replication_0070_2'
        self.slot_name_upper = 'SLOT_logical_replication_0070'
        self.slot_name_invalid = 'slot_logical_replication_0070_*'
        self.slot_name_63chars = \
            'slot_logical_replication_0070_slot_logical_replication_0070_slo'
        self.slot_name_64chars = \
            'slot_logical_replication_0070_slot_logical_replication_0070_slot'
        self.config_slot_log = 'enable_slot_log'
        self.slot_log_default = self.common.show_param(self.config_slot_log)
        self.config_wal_level = 'wal_level'
        self.wal_level_default = self.common.show_param(self.config_wal_level)

    def test_logical_replication(self):
        text = "-----step1:修改参数;expect:修改参数成功-----"
        self.logger.info(text)
        if 'on' != self.slot_log_default:
            guc_result = primary_sh.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_slot_log}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.common.show_param(self.config_slot_log)
        self.assertEquals(show_msg, 'on', '执行失败' + text)

        if 'logical' != self.wal_level_default:
            guc_result = primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)
            restart_msg = primary_sh.restart_db_cluster()
            self.logger.info(restart_msg)
            self.assertTrue(restart_msg, '重启失败')
        show_msg = self.common.show_param(self.config_wal_level)
        self.assertEquals(show_msg, 'logical', '执行失败' + text)

        text = "-----step2:配置初始用户白名单;expect:配置成功-----"
        self.logger.info(text)
        param_1 = f'local replication {self.primary_node.ssh_user}  trust'
        param_2 = f'host  replication {self.primary_node.ssh_user} ' \
            f'127.0.0.1/32 trust'
        param_3 = f'host  replication {self.primary_node.ssh_user} ' \
            f'::1/128 trust'
        result_1 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_1)
        result_2 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_2)
        result_3 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text)

        text = "-----step3:创建逻辑复制槽;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f"select pg_create_logical_replication_slot(" \
            f"'{self.slot_name_1}', 'mppdb_decoding');" \
            f"select * from pg_get_replication_slots() " \
            f"where slot_name like '%{self.slot_name_1}%';"
        sql_msg = primary_sh.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name_1, sql_msg.splitlines()[-2],
                      '执行失败:' + text)

        text = "-----step4:在主机上执行此函数，" \
               "expect:只有存在的slotname执行成功，其他的合理报错-----"
        self.logger.info(text)
        text = "-----step4.1:在主机上执行此函数，命名不符合规范的" \
               "expect:合理报错-----"
        self.logger.info(text)
        peek_cmd = f"select pg_logical_slot_peek_binary_changes" \
            f"('{self.slot_name_invalid}', '', 3,'include-xids', '1');" \
            f"select pg_logical_slot_peek_binary_changes" \
            f"('{self.slot_name_upper}', '', 3,'include-xids', '1');"
        peek_msg = primary_sh.execut_db_sql(peek_cmd, dbname='postgres')
        self.logger.info(peek_msg)
        self.assertEquals(2, peek_msg.count(self.constant.invalid_character),
                          '执行失败:' + text)

        text = "-----step4.2:在主机上执行此函数，命名长度为临界值," \
               "expect:不校验长度，报错为not exist-----"
        self.logger.info(text)
        peek_cmd = f"select pg_logical_slot_peek_binary_changes" \
            f"('{self.slot_name_63chars}', '',3,'include-xids', '1');" \
            f"select pg_logical_slot_peek_binary_changes" \
            f"('{self.slot_name_64chars}', '',3,'include-xids', '1');"
        peek_msg = primary_sh.execut_db_sql(peek_cmd, dbname='postgres')
        self.logger.info(peek_msg)
        self.assertEquals(2, peek_msg.count("not exist"), '执行失败:' + text)

        text = "-----step4.3:在主机上执行此函数，不存在的slotname," \
               "expect:合理报错:not exist-----"
        self.logger.info(text)
        peek_cmd = f"select pg_logical_slot_peek_binary_changes" \
            f"('{self.slot_name_2}', '', 3,'include-xids', '1');"
        peek_msg = primary_sh.execut_db_sql(peek_cmd, dbname='postgres')
        self.logger.info(peek_msg)
        self.assertIn("ERROR" and "not exist", peek_msg, '执行失败:' + text)

        text = "-----step4.4:在主机上执行此函数，slotname存在;expect:成功-----"
        self.logger.info(text)
        peek_cmd = f"select pg_logical_slot_peek_binary_changes" \
            f"('{self.slot_name_1}', '', 3,'include-xids', '1');"
        peek_msg = primary_sh.execut_db_sql(peek_cmd, dbname='postgres')
        self.logger.info(peek_msg)
        self.assertTrue("ERROR" not in peek_msg and
                        "(0 rows)" in peek_msg, '执行失败:' + text)

        text = "-----step5:在备机上执行此函数;" \
               "expect:只有存在的slotname执行成功，其他的合理报错-----"
        self.logger.info(text)
        text = "-----step5.1:在备机上执行此函数，命名不符合规范的" \
               "expect:合理报错-----"
        self.logger.info(text)
        peek_cmd = f"select pg_logical_slot_peek_binary_changes" \
            f"('{self.slot_name_invalid}', '', 3,'include-xids', '1');" \
            f"select pg_logical_slot_peek_binary_changes" \
            f"('{self.slot_name_upper}', '', 3,'include-xids', '1');"
        peek_msg = self.sh_standby.execut_db_sql(peek_cmd, dbname='postgres')
        self.logger.info(peek_msg)
        self.assertEquals(2, peek_msg.count(self.constant.invalid_character),
                          '执行失败:' + text)

        text = "-----step5.2:在备机上执行此函数，命名超过64个字符的" \
               "expect:合理报错-----"
        self.logger.info(text)
        peek_cmd = f"select pg_logical_slot_peek_binary_changes" \
            f"('{self.slot_name_63chars}', '',3,'include-xids', '1');" \
            f"select pg_logical_slot_peek_binary_changes" \
            f"('{self.slot_name_64chars}', '',3,'include-xids', '1');"
        peek_msg = self.sh_standby.execut_db_sql(peek_cmd, dbname='postgres')
        self.logger.info(peek_msg)
        self.assertEquals(2, peek_msg.count("not exist"), '执行失败:' + text)

        text = "-----step5.3:在备机上执行此函数，不存在的slotname" \
               "expect:合理报错-----"
        self.logger.info(text)
        peek_cmd = f"select pg_logical_slot_peek_binary_changes" \
            f"('{self.slot_name_2}', '', 3,'include-xids', '1');"
        peek_msg = self.sh_standby.execut_db_sql(peek_cmd, dbname='postgres')
        self.logger.info(peek_msg)
        self.assertIn("ERROR" and "not exist", peek_msg, '执行失败:' + text)

        text = "-----step5.4:在备机上执行此函数，slotname存在;expect:成功-----"
        self.logger.info(text)
        peek_cmd = f"select pg_logical_slot_peek_binary_changes" \
            f"('{self.slot_name_1}', '', 3,'include-xids', '1');"
        peek_msg = self.sh_standby.execut_db_sql(peek_cmd, dbname='postgres')
        self.logger.info(peek_msg)
        self.assertTrue("ERROR" not in peek_msg and
                        "(0 rows)" in peek_msg, '执行失败:' + text)

    def tearDown(self):
        text_1 = "-----step6:删除逻辑复制槽;expect:删除成功-----"
        self.logger.info(text_1)
        sql_cmd = f"select pg_drop_replication_slot('{self.slot_name_1}');" \
            f"select * from pg_get_replication_slots() " \
            f"where slot_name like '%{self.slot_name_1}%';"
        sql_msg = primary_sh.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)

        text_2 = "-----step7:恢复参数，恢复pg_hba.conf文件，" \
                 "expect:恢复成功-----"
        self.logger.info(text_2)
        slot_log_current = self.common.show_param(self.config_slot_log)
        if slot_log_current != self.slot_log_default:
            primary_sh.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_slot_log}={self.slot_log_default}')
        show_slot_log = self.common.show_param(self.config_slot_log)

        param_1 = f'local replication {self.primary_node.ssh_user} '
        param_2 = f'host replication {self.primary_node.ssh_user} 127.0.0.1/32 '
        param_3 = f'host  replication {self.primary_node.ssh_user} ::1/128 '
        result_1 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_1)
        result_2 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_2)
        result_3 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_3)

        wal_level_current = self.common.show_param(self.config_wal_level)
        if wal_level_current != self.wal_level_default:
            primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}={self.wal_level_default}')
            restart_msg = primary_sh.restart_db_cluster()
            self.logger.info(restart_msg)
        show_wal_level = self.common.show_param(self.config_wal_level)

        self.assertIn("(0 rows)", sql_msg, '执行失败:' + text_1)
        self.assertEquals(show_slot_log, self.slot_log_default,
                          '执行失败' + text_2)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text_2)
        self.assertEquals(show_wal_level, self.wal_level_default,
                          '执行失败' + text_2)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")
